package ds_industry_2025.industry.gy_08.T5

import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.classification.RandomForestClassifier
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
import org.apache.spark.ml.feature.VectorAssembler
import org.apache.spark.ml.param.ParamMap
import org.apache.spark.ml.tuning.{CrossValidator, CrossValidatorModel, ParamGridBuilder}
import org.apache.spark.sql.SparkSession

/*
      1、根据子任务一的结果，建立随机森林（随机森林相关参数可自定义，不做限制），使用子任务一的结果训练随机森林模型，然后再将
      dwd.fact_machine_learning_data_test（该表字段含义与dwd.fact_machine_learning_data表相同，machine_record_state列
      值为空，表结构自行查看）转成向量，预测其是否报警将结果输出到MySQL数据库shtd_industry中的ml_result表中（表结构如下）。
      在Linux的MySQL命令行中查询出machine_record_id为1、8、20、28和36的5条数据，将SQL语句复制并粘贴至客户端桌面【Release\
      任务C提交结果.docx】中对应的任务序号下，将执行结果截图粘贴至客户端桌面【Release\任务C提交结果.docx】中对应的任务序号下。
ml_result表结构：
字段	类型	中文含义	备注
machine_record_id	int	主键
machine_record_state	double	设备状态	报警为1，其他状态则为0
 */
object t1 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("随机森林")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()



    import spark.implicits._

    //  todo 读取特征工程的结果
    val data = spark.table("dwd.fact_machine_learning_data")

    //  todo 拿到特征字段   之所以是从3开始是因为前三个字段不是特征  ，然后最后五个也不需要，这些截取的字段就是从xml解析出来的
    //  todo 其中下面更改字段名，这一步非常重要，因为在监督学习中，模型需要知道哪个字段是目标变量(即我们要预测的变量)
    //  todo 这个目标变量通常被称为标签(label)
    val cols:Array[String] = data.columns.slice(3, data.columns.length - 5)
    val Array(ranting,test) = data.na.fill(0.0, cols)
      .withColumnRenamed("machine_record_state", "label") //  把状态字段改为label
      .randomSplit(Array(0.8, 0.2))

    //  todo 定义将特征全部转化为特征向量的vectorAssembler
    val vectorAssembler = new VectorAssembler()
      .setInputCols(cols)
      .setOutputCol("features")

    //  todo 创建一个随机森林分类器
    val classifier = new RandomForestClassifier()

      //  todo 定义机器学习流水线
      val pipeline = new Pipeline()
        .setStages(Array(vectorAssembler,classifier))

    //  通过构建一个超参数网格和使用交叉验证的方式，系统地评估不同的随机森林模型参数组合，以找到最佳配置，从而提高模型的预测性能。
    /*
        创建超参数网格
    val paramMaps:Array[ParamMap]=new ParamGridBuilder()
      .addGrid(classifier.numTrees,Array(10,20,30))
      .addGrid(classifier.maxDepth,Array(5,10,15))
      .addGrid(classifier.impurity,Array("entropy","gini"))
      .build()
    ParamGridBuilder: 用于构建超参数的网格。
    addGrid: 向参数网格中添加不同的参数组合。每个addGrid方法指定一个要调整的参数和一组可能的值。
    classifier.numTrees: 随机森林中的树的数量，这里尝试10、20和30棵树。
    classifier.maxDepth: 每棵树的最大深度，尝试深度为5、10和15。
    classifier.impurity: 用于衡量不纯度的标准，这里使用"entropy"（信息增益）和"gini"（基尼不纯度）。
    build(): 生成参数组合的数组（Array[ParamMap]），每个组合将用于训练模型。
     */
    //   param:参数         grid:网格     builder:构造器
    val paramMaps: Array[ParamMap] = new ParamGridBuilder()
      .addGrid(classifier.numTrees, Array(10, 20, 30))
      .addGrid(classifier.maxDepth, Array(5, 10, 15))
      .addGrid(classifier.impurity, Array("entropy", "gini")) //  impurity:杂质
      .build()


    /*
        CrossValidator: Spark中的交叉验证类，用于评估模型的性能并找到最佳的超参数组合。
    setEstimatorParamMaps(paramMaps): 设置要测试的参数组合，即在第一步构建的paramMaps。
    setEvaluator(new MulticlassClassificationEvaluator()): 设置评估器，这里使用多类分类评估器，评估模型的分类性能（如准确率、召回率等）。
    setEstimator(pipeline): 设置要训练的模型管道，这个管道包括特征提取和分类器。
    setParallelism(3): 指定并行度，表示在交叉验证过程中，可以同时运行3个模型训练任务，以提高效率。
     */
    //  cross:交叉          Validator:验证器
    val crossValidator: CrossValidator = new CrossValidator()
      .setEstimatorParamMaps(paramMaps)
      .setEvaluator(new MulticlassClassificationEvaluator())
      .setEstimator(pipeline)
      .setParallelism(4)

    //  训练模型并且保存
    val model: CrossValidatorModel = crossValidator.fit(ranting)

    //  读取需要预测的数据
//    val testData = spark.table("dwd.fact_machine_learning_data_test")

    //  使用训练好的模型进行预测
    // 由于在交叉验证里面选择了流水线，所以测试数据不需要再去转化特征向量，自己会做这样的流程
    val predictions = model.transform(test)

    //  选择需要的列      prediction:预测
    val result = predictions.select("machine_record_id", "prediction")
      .withColumnRenamed("prediction", "machine_record_state")
      .show


    //  select * from ml_result where machine_record_id in(14747628,14747632,14747645,14747630,14748618,14748623);

    spark.close()
  }
}
